package com.cobra.wc

import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

//sc.textFile("/opt/module/data/word.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect().foreach(println);
object Spark05_WordCount {
  //groupBy
  def wordcount1(sc: SparkContext)  = {
    val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    val words = rdd.flatMap(_.split(" "))
    val group = words.groupBy(word=>word)
    val wordCount = group.mapValues(iter => iter.size)
  }
  //groupByKey
  def wordcount2(sc: SparkContext)  = {
    val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    val words = rdd.flatMap(_.split(" "))
    val wordOne = words.map((_, 1))
    val group = wordOne.groupByKey()
    val wordCount = group.mapValues(iter => iter.size)
  }

  //reduceByKey
  def wordcount3(sc: SparkContext)  = {
    val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    val words = rdd.flatMap(_.split(" "))
    val wordOne = words.map((_, 1))
    val wordCount = wordOne.reduceByKey(_ + _)
  }

  //aggregateByKey
  def wordcount4(sc: SparkContext)  = {
    val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    val words = rdd.flatMap(_.split(" "))
    val wordOne = words.map((_, 1))
    val wordCount = wordOne.aggregateByKey(0)(_+_,_+_)
  }
  //foldByKey
  def wordcount5(sc: SparkContext)  = {
    val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    val words = rdd.flatMap(_.split(" "))
    val wordOne = words.map((_, 1))
    val wordCount = wordOne.foldByKey(0)(_+_)
  }
  //combineByKey
  def wordcount6(sc: SparkContext)  = {
    val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    val words = rdd.flatMap(_.split(" "))
    val wordOne = words.map((_, 1))
    val wordCount = wordOne.combineByKey(
      v=>v,
      (x:Int,y:Int)=> x+y,
      (x:Int,y:Int)=> x+y
    )
  }
  //countByKey
  def wordcount7(sc: SparkContext)  = {
    val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    val words = rdd.flatMap(_.split(" "))
    val wordOne = words.map((_, 1))
    val wordCount = wordOne.countByKey()
  }

  //countByKey
  def wordcount8(sc: SparkContext)  = {
    val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    val words = rdd.flatMap(_.split(" "))
    val wordCount = words.countByValue()
  }


  //reduce
  def wordcount9(sc: SparkContext)  = {
    val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
    val words = rdd.flatMap(_.split(" "))
    val mapWord = words.map(
      word => {
        mutable.Map[String, Long]((word, 1))
      }
    )
    mapWord.foreach(println)
    println("=========================")
    val wordCount = mapWord.reduce(
      (map1, map2) => {
        map2.foreach{
          case(word,count)=>{
            val newCount = map1.getOrElse(word,0L)+count
            map1.update(word,newCount)
          }
        }
        println(map1)
        println("=========================")
        map1
      }
    )
    println(wordCount)
  }





  def main(args: Array[String]): Unit = {
    //建立和Spark框架的连接
    val conf = new SparkConf().setMaster("local").setAppName("WordCount5")
    val sc = new SparkContext(conf)

    wordcount9(sc)

    sc.stop()
  }
}
